package cn.hy.dataSync.utils;


import cn.hy.dataSync.config.DataBaseType;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.commons.lang3.tuple.Triple;

import java.math.BigInteger;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;


@Slf4j
public final class RdbUtils {

    private static final int SOCKET_TIMEOUT_SECOND = 3600;

    private static final ThreadLocal<ExecutorService> RS_ASYNC_EXECUTORS =
            ThreadLocal.withInitial(() -> Executors.newFixedThreadPool(1, new ThreadFactoryBuilder()
                    .setNameFormat("rsAsyncExecutors-%d")
                    .setDaemon(true)
                    .build()));

    private RdbUtils() {
    }

    public static List<String> splitSingleTable(String jdbcUrl,
                                                String username,
                                                String password,
                                                String table,
                                                String column,
                                                String splitPkName,
                                                String where,
                                                int adviceNum) {
        return splitSingleTable(DataBaseType.parseDataBaseTypeFromJdbcUrl(jdbcUrl),
                jdbcUrl, username, password, table, column, splitPkName, where, adviceNum);
    }

    public static List<String> splitSingleTable(DataBaseType dataBaseType,
                                                String jdbcUrl,
                                                String username,
                                                String password,
                                                String table,
                                                String column,
                                                String splitPkName,
                                                String where,
                                                int adviceNum) {

        Connection conn = getConnection(dataBaseType, jdbcUrl, username, password);
        try {
            return splitSingleTable(dataBaseType, conn, table, column, splitPkName, where, adviceNum);
        } finally {
            close(conn);
        }
    }

    public static List<String> splitSingleTable(DataBaseType dataBaseType,
                                                Connection conn,
                                                String table,
                                                String column,
                                                String splitPkName,
                                                String where,
                                                int adviceNum) {
        boolean hasWhere = StringUtils.isNotBlank(where);

        //三元组 (minId, maxId , DataType)
        Triple<Object, Object, Integer> minMaxPk = getPkRange(conn, table, splitPkName, where);
        if (null == minMaxPk) {
            throw new RuntimeException(
                    "根据切分主键切分表失败. 仅支持切分主键为一个,并且类型为整数或者字符串类型");
        }

        // 有一个为空
        // 这个情况存在吗 ？
        if (null == minMaxPk.getLeft() || null == minMaxPk.getRight()) {
            List<String> singleQuerySql = new ArrayList<>();
            singleQuerySql.add(buildQuerySql(column, table, where));
            return singleQuerySql;
        }

        // 范围SQL
        List<String> rangeList;

        if (isStringType(minMaxPk.getRight())) {
            rangeList = RangeSplitUtils.splitAndWrap(
                    String.valueOf(minMaxPk.getLeft()),
                    String.valueOf(minMaxPk.getMiddle()),
                    adviceNum, splitPkName, "'", dataBaseType);
        } else if (isLongType(minMaxPk.getRight())) {
            rangeList = RangeSplitUtils.splitAndWrap(
                    new BigInteger(minMaxPk.getLeft().toString()),
                    new BigInteger(minMaxPk.getMiddle().toString()),
                    adviceNum, splitPkName);
        } else {
            throw new RuntimeException("您配置的切分主键(splitPk) 类型不支持. 仅支持切分主键为一个,并且类型为整数或者字符串类型");
        }

        String tempQuerySql;
        // 完成切分的SQL
        List<String> allQuerySql = new ArrayList<>();

        if (Objects.nonNull(rangeList) && !rangeList.isEmpty()) {
            for (String range : rangeList) {
                tempQuerySql = buildQuerySql(column, table, where)
                        + (hasWhere ? " and " : " where ") + range;

                allQuerySql.add(tempQuerySql);
            }
        } else {
            tempQuerySql = buildQuerySql(column, table, where)
                    + (hasWhere ? " and " : " where ")
                    + String.format(" %s is not null", splitPkName);
            allQuerySql.add(tempQuerySql);
        }

        // 这里是干嘛的？
        // deal pk is null
        tempQuerySql = buildQuerySql(column, table, where)
                + (hasWhere ? " and " : " where ")
                + String.format(" %s is null", splitPkName);

        allQuerySql.add(tempQuerySql);

        return allQuerySql;
    }

    private static Triple<Object, Object, Integer> getPkRange(DataBaseType dataBaseType,
                                                              String jdbcUrl,
                                                              String username,
                                                              String password,
                                                              String table,
                                                              String splitPkName,
                                                              String where) {
        Connection conn = getConnection(dataBaseType, jdbcUrl, username, password);
        try {
            return getPkRange(conn, table, splitPkName, where);
        } finally {
            close(conn);
        }
    }

    private static Triple<Object, Object, Integer> getPkRange(Connection conn,
                                                              String table,
                                                              String splitPkName,
                                                              String where) {
        return checkSplitPk(conn, genPkSql(splitPkName, table, where));
    }

    private static Triple<Object, Object, Integer> checkSplitPk(Connection conn,
                                                                String pkRangeSql) {
        ResultSet rs = null;
        Triple<Object, Object, Integer> minMaxPk = null;
        try {
            try {
                rs = query(conn, pkRangeSql);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
            ResultSetMetaData rsMetaData = rs.getMetaData();
            if (isPkTypeValid(rsMetaData)) {
                int columnType = rsMetaData.getColumnType(1);
                if (isStringType(columnType) || isLongType(columnType)) {
                    while (asyncResultSetNext(rs)) {
                        minMaxPk = new ImmutableTriple<>(rs.getString(1), rs.getString(2), columnType);
                    }
                } else {
                    throw new RuntimeException(
                            "您配置的切分主键(splitPk)有误. 因为您配置的切分主键(splitPk) 类型不支持. 仅支持切分主键为一个,并且类型为整数或者字符串类型");
                }
            } else {
                throw new RuntimeException(
                        "您配置的切分主键(splitPk)有误. 因为您配置的切分主键(splitPk) 类型不支持. 仅支持切分主键为一个,并且类型为整数或者字符串类型");
            }
        } catch (Exception e) {
            throw new RuntimeException("尝试切分表发生错误. 请检查您的配置并作出修改.", e);
        } finally {
            close(rs);
        }
        return minMaxPk;
    }

    private static boolean isPkTypeValid(ResultSetMetaData rsMetaData) {
        boolean ret = false;
        try {
            int minType = rsMetaData.getColumnType(1);
            int maxType = rsMetaData.getColumnType(2);

            boolean isNumberType = isLongType(minType);

            boolean isStringType = isStringType(minType);

            if (minType == maxType && (isNumberType || isStringType)) {
                ret = true;
            }
        } catch (Exception e) {
            throw new RuntimeException("获取切分主键(splitPk)字段类型失败");
        }
        return ret;
    }

    private static boolean isLongType(int type) {
        return type == Types.BIGINT || type == Types.INTEGER
                || type == Types.SMALLINT || type == Types.TINYINT;
    }

    private static boolean isStringType(int type) {
        return type == Types.CHAR || type == Types.NCHAR
                || type == Types.VARCHAR || type == Types.LONGVARCHAR
                || type == Types.NVARCHAR;
    }

    /**
     *  生成id 范围的SQL
     * @param splitPk id
     * @param table 表
     * @param where where 条件
     * @return SQL
     */
    private static String genPkSql(String splitPk, String table, String where) {
        String minMaxTemplate = "select min(%s),max(%s) from %s";
        String pkRangeSql = String.format(minMaxTemplate, splitPk, splitPk, table);
        if (StringUtils.isNotBlank(where)) {
            pkRangeSql = String.format("%s where (%s and %s is not null)",
                    pkRangeSql, where, splitPk);
        }
        return pkRangeSql;
    }

    private static String buildQuerySql(String column,
                                        String table,
                                        String where) {
        String querySql;

        if (StringUtils.isBlank(where)) {
            querySql = String.format("select %s from %s ", column, table);
        } else {
            querySql = String.format("select %s from %s where (%s)", column, table, where);
        }

        return querySql;
    }

    public static String getWriteSql(List<String> columnHolders,
                                     List<String> valueHolders,
                                     String writeMode,
                                     DataBaseType dataBaseType) {
        boolean isWriteModeLegal = writeMode.trim().toLowerCase().startsWith("insert")
                || writeMode.trim().toLowerCase().startsWith("replace")
                || writeMode.trim().toLowerCase().startsWith("update");

        if (!isWriteModeLegal) {
            throw new RuntimeException(String.format(
                    "您所配置的 writeMode:%s 错误. 因为目前仅支持 replace, update 或 insert 方式. 请检查您的配置并作出修改", writeMode));
        }
        String writeDataSqlTemplate;
        if (dataBaseType == DataBaseType.MySql && writeMode.trim().toLowerCase().startsWith("update")) {
            StringBuilder stringBuilder = new StringBuilder();
            stringBuilder.append("update %s set ");
            for (int i = 0; i < columnHolders.size(); i++) {
                stringBuilder.append(columnHolders.get(i));
                stringBuilder.append("=");
                stringBuilder.append(valueHolders.get(i));
                stringBuilder.append(",");
            }
            stringBuilder.setLength(stringBuilder.length() - 1);
            stringBuilder.append(" where %s = ");
            stringBuilder.append(valueHolders.get(0));
            writeDataSqlTemplate = stringBuilder.toString();
        } else {
            if (writeMode.trim().toLowerCase().startsWith("update")) {
                writeMode = "replace";
            }
            writeDataSqlTemplate = writeMode +
                    " into %s (" + StringUtils.join(columnHolders, ",") +
                    ") values(" + StringUtils.join(valueHolders, ",") + ")";
        }
        return writeDataSqlTemplate;
    }

    public static String onDuplicateKeyUpdateString(List<String> columnHolders) {
        if (columnHolders == null || columnHolders.size() < 1) {
            return "";
        }
        StringBuilder sb = new StringBuilder();
        sb.append(" on duplicate key update ");
        boolean first = true;
        for (String column : columnHolders) {
            if (!first) {
                sb.append(",");
            } else {
                first = false;
            }
            sb.append(column);
            sb.append("=values(");
            sb.append(column);
            sb.append(")");
        }

        return sb.toString();
    }

    public static boolean asyncResultSetNext(final ResultSet resultSet) {
        return asyncResultSetNext(resultSet, SOCKET_TIMEOUT_SECOND);
    }

    public static boolean asyncResultSetNext(final ResultSet resultSet, int timeout) {
        Future<Boolean> future = RS_ASYNC_EXECUTORS.get().submit(resultSet::next);
        try {
            return future.get(timeout, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new RuntimeException("异步获取ResultSet失败", e);
        }
    }

    public static String getTableDdl(Connection conn, String table) {
        List<String> tables = new ArrayList<>();
        tables.add(table);
        try {
            List<String> result =
                    com.alibaba.druid.util.MySqlUtils.getTableDDL(conn, tables);
            return result.size() != 0 ? result.get(0) : null;
        } catch (SQLException se) {
            throw new RuntimeException(se);
        }
    }

    public static List<String> getTableColumns(String jdbcUrl,
                                               String username,
                                               String password,
                                               String table) {
        Connection conn = getConnection(DataBaseType.parseDataBaseTypeFromJdbcUrl(jdbcUrl), jdbcUrl, username, password);
        try {
            return getTableColumns(conn, table);
        } finally {
            close(conn);
        }
    }

    public static List<String> getTableColumns(Connection conn, String table) {
        List<String> columns = new ArrayList<>();
        Statement statement = null;
        ResultSet rs = null;
        try {
            statement = conn.createStatement();
            String queryColumnSql = String.format("select * from %s where 1=2", table);
            rs = statement.executeQuery(queryColumnSql);
            ResultSetMetaData rsMetaData = rs.getMetaData();
            for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
                columns.add(rsMetaData.getColumnName(i + 1));
            }
        } catch (SQLException se) {
            throw new RuntimeException(se.getMessage());
        } finally {
            close(statement, rs);
        }
        return columns;
    }

    /**
     * @return Left:ColumnName Middle:ColumnType Right:ColumnTypeName
     */
    public static Triple<List<String>, List<Integer>, List<String>> getColumnMetaData(
            DataBaseType dataBaseType,
            String jdbcUrl,
            String username,
            String password,
            String tableName,
            String column) {
        Connection conn = null;
        try {
            conn = getConnection(dataBaseType, jdbcUrl, username, password);
            return getColumnMetaData(conn, tableName, column);
        } finally {
            close(conn);
        }
    }

    /**
     * @return Left:ColumnName Middle:ColumnType Right:ColumnTypeName
     */
    public static Triple<List<String>, List<Integer>, List<String>> getColumnMetaData(Connection conn, String tableName, String column) {
        Statement statement = null;
        ResultSet rs = null;

        Triple<List<String>, List<Integer>, List<String>> columnMetaData =
                new ImmutableTriple<>(
                        new ArrayList<>(),
                        new ArrayList<>(),
                        new ArrayList<>()
                );
        try {
            statement = conn.createStatement();
            String queryColumnSql = String.format("select %s from %s where 1=2", column, tableName);
            rs = statement.executeQuery(queryColumnSql);
            ResultSetMetaData rsMetaData = rs.getMetaData();
            for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {
                columnMetaData.getLeft().add(rsMetaData.getColumnName(i + 1));
                columnMetaData.getMiddle().add(rsMetaData.getColumnType(i + 1));
                columnMetaData.getRight().add(rsMetaData.getColumnTypeName(i + 1));
            }
            return columnMetaData;

        } catch (SQLException se) {
            throw new RuntimeException(String.format("获取表:%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", tableName), se);
        } finally {
            close(statement, rs);
        }
    }

    public static Connection getConnection(final DataBaseType dataBaseType,
                                           String jdbcUrl,
                                           String username,
                                           String password) {
        try {
            return RetryUtils.executeWithRetry(() -> connect(
                    dataBaseType,
                    jdbcUrl,
                    username,
                    password), 9, 1000L, true);
        } catch (Exception e) {
            throw new RuntimeException(
                    String.format("数据库连接失败. 因为根据您配置的连接信息:%s获取数据库连接失败. 请检查您的配置并作出修改.", jdbcUrl), e);
        }
    }

    public static Connection getConnectionWithoutRetry(final DataBaseType dataBaseType,
                                                       String jdbcUrl,
                                                       String username,
                                                       String password) {
        return connect(
                dataBaseType,
                jdbcUrl,
                username,
                password
        );
    }

    private static synchronized Connection connect(final DataBaseType dataBaseType,
                                                   String jdbcUrl,
                                                   String username,
                                                   String password) {
        try {
            Class.forName(dataBaseType.getDriverClass());
            return DriverManager.getConnection(jdbcUrl, username, password);
        } catch (ClassNotFoundException | SQLException ce) {
            throw new RuntimeException(ce);
        }
    }

    public static ResultSet query(Connection conn, String sql)
            throws SQLException {
        // 默认3600 s 的query Timeout
        return query(conn, sql, SOCKET_TIMEOUT_SECOND);
    }

    public static ResultSet query(Connection conn, String sql, int queryTimeout)
            throws SQLException {
        // make sure autocommit is off
        conn.setAutoCommit(false);
        Statement stmt = conn.createStatement(
                ResultSet.TYPE_FORWARD_ONLY,
                ResultSet.CONCUR_READ_ONLY);
        stmt.setQueryTimeout(queryTimeout);
        stmt.setFetchSize(10000);
        return query(stmt, sql);
    }

    public static ResultSet query(Statement stmt, String sql)
            throws SQLException {
        return stmt.executeQuery(sql);
    }

    public static void executeSqlWithoutResultSet(Statement stmt, String sql)
            throws SQLException {
        stmt.execute(sql);
    }

    public static void close(Connection conn, Statement stmt, ResultSet rs) {
        close(rs);
        close(stmt);
        close(conn);
    }

    public static void close(Connection conn, Statement stmt) {
        close(stmt);
        close(conn);
    }

    public static void close(Statement stmt, ResultSet rs) {
        close(rs);
        close(stmt);
    }

    public static void close(ResultSet rs) {
        if (Objects.nonNull(rs)) {
            try {
                rs.close();
            } catch (SQLException unused) {
                //
            }
        }
    }

    public static void close(Statement stmt) {
        if (Objects.nonNull(stmt)) {
            try {
                stmt.close();
            } catch (SQLException unused) {
                //
            }
        }
    }

    public static void close(Connection conn) {
        if (Objects.nonNull(conn)) {
            try {
                conn.close();
            } catch (SQLException unused) {
                //
            }
        }
    }
}
